package com.pisolution.scheduler.runtime.process;

import com.pisolution.scheduler.core.utils.JsonUtils;
import com.pisolution.scheduler.datasource.command.Command;
import com.pisolution.scheduler.datasource.command.CommandMapper;
import com.pisolution.scheduler.datasource.command.CommandType;
import com.pisolution.scheduler.datasource.command.ErrorCommand;
import com.pisolution.scheduler.datasource.command.ErrorCommandMapper;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.datasource.instance.Flag;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.datasource.instance.TaskInstanceMapper;
import com.pisolution.scheduler.datasource.instance.WarningType;
import com.pisolution.scheduler.datasource.instance.WorkflowInstance;
import com.pisolution.scheduler.datasource.instance.WorkflowInstanceMapper;
import com.pisolution.scheduler.datasource.workflow.Workflow;
import com.pisolution.scheduler.datasource.workflow.WorkflowMapper;

import com.google.gson.reflect.TypeToken;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;

/**
 * 辅助运行时节点组件之间的数据交互 主要负责以下内容.
 *
 * <ul>
 *     <li>实例的失效转移</li>
 *     <li>系统控制指令集的处理</li>
 * </ul>
 *
 * @author sora
 * @version 0.1.0
 */
@Component
public class ProcessService {
    private static final Logger LOGGER = LoggerFactory.getLogger(ProcessService.class);

    // -------------------------------------------------------------------------------------------------

    /** 尚未运行完成的工作流或任务节点的当前所有可能状态. */
    private static final int[] STATUS = new int[]{
        ExecutionState.SUBMITTED_SUCCESS.ordinal(),
        ExecutionState.RUNNING.ordinal(),
        ExecutionState.READY_PAUSE.ordinal(),
        ExecutionState.READY_STOP.ordinal()
    };

    // -------------------------------------------------------------------------------------------------

    /** 用于与工作流实例进行交互. */
    private final WorkflowInstanceMapper workflowInstance;

    /** 用于与系统控制指令进行交互. */
    private final CommandMapper command;

    /** 用于与任务实例进行交互. */
    private final TaskInstanceMapper taskInstance;

    /** 用于与工作流定义进行交互. */
    private final WorkflowMapper workflow;

    /** 用于与执行错误的系统控制指令进行交互. */
    private final ErrorCommandMapper errorCommand;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化该辅助工具.
     *
     * @param workflowInstance 用于与工作流实例进行交互.
     * @param command 用于与系统控制指令进行交互.
     * @param taskInstance 用于与任务实例进行交互.
     * @param workflow 用于与工作流定义进行交互.
     * @param errorCommand 用于与执行错误的系统控制指令进行交互.
     */
    public ProcessService(
        final WorkflowInstanceMapper workflowInstance,
        final CommandMapper command,
        final TaskInstanceMapper taskInstance,
        final WorkflowMapper workflow,
        final ErrorCommandMapper errorCommand
    ) {
        this.workflowInstance = workflowInstance;
        this.command = command;
        this.taskInstance = taskInstance;
        this.workflow = workflow;
        this.errorCommand = errorCommand;
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (基础查询)
    // -------------------------------------------------------------------------------------------------

    /**
     * 查询符合给定工作流实例标识的工作流实例对象.
     *
     * @param instanceId 待查询工作流实例标识.
     *
     * @return 符合该限定条件的工作流实例对象.
     */
    public WorkflowInstance findWorkflowInstanceById(final int instanceId) {
        return this.workflowInstance.selectById(instanceId);
    }

    /**
     * 查询给定工作流实例下所有任务实例对象.
     *
     * @param instanceId 待查询工作流实例标识.
     *
     * @return 符合该限定条件的任务实例对象.
     */
    public List<TaskInstance> findTaskInstancesByWorkflowInstanceId(final int instanceId) {
        return this.taskInstance.queryByWorkflowInstanceId(instanceId);
    }

    /**
     * 查询符合给定任务实例标识的任务实例.
     *
     * @param instanceId 待查询任务实例标识.
     *
     * @return 符合该限定条件的任务实例对象.
     */
    @Nullable
    public TaskInstance findTaskInstanceById(final int instanceId) {
        return this.taskInstance.selectById(instanceId);
    }

    /**
     * 保存任务实例对象至数据库中 如果该任务实例不存在 则创建该实例 否则更新原有实例对象.
     *
     * @param instance 待保存任务实例对象.
     *
     * @return 是否保存成功.
     */
    @SuppressWarnings("UnusedReturnValue")
    public boolean saveTaskInstance(final TaskInstance instance) {
        if (instance.getId() != 0) {
            return this.taskInstance.updateById(instance) > 0;
        }

        return this.taskInstance.insert(instance) > 0;
    }

    /**
     * 保存任务实例对象至数据库中 如果该工作流实例不存在 则创建该实例 否则更新原有实例对象.
     *
     * @param instance 待保存工作流实例对象.
     */
    public void saveWorkflowInstance(final WorkflowInstance instance) {
        if (instance.getId() == 0) {
            this.workflowInstance.insert(instance);

            return;
        }

        this.workflowInstance.updateById(instance);
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (失效转移)
    // -------------------------------------------------------------------------------------------------

    /**
     * 查询所有需要进行失效转移的工作流实例.
     *
     * @param host 管理工作流实例的管理节点主机地址 如果未指定 则忽略主机地址该限定条件.
     *
     * @return 需要进行失效转移的工作流实例.
     */
    public List<WorkflowInstance> queryWorkflowInstanceNeedFailover(@Nullable final String host) {
        return this.workflowInstance.queryByHostAndState(host, STATUS);
    }

    /**
     * 处理需要进行失效转移的工作流实例.
     *
     * @param instance 需要恢复的已失效工作流实例.
     */
    @Transactional(rollbackFor = Exception.class)
    public void processInstanceNeedFailover(final WorkflowInstance instance) {
        LOGGER.info("Set the host of the workflow instance {} to NULL.", instance.getId());

        // 1. 清空进程实例的运行主机信息.
        instance.setHost("NULL");
        this.workflowInstance.updateById(instance);

        // 2. 构建系统指令以完成该工作流实例的失效转移.
        LOGGER.info("Create the failover command for the workflow instance {}.", instance.getId());

        final Command command = new Command();

        command.setWorkflowId(instance.getWorkflowId());
        command.setType(CommandType.RECOVER_FAULT_TOLERANT);
        command.setParams(String.format("{\"workflowInstanceId\": %d}", instance.getId()));

        // 3. 将该系统指令更新到待处理指令集中.
        this.createCommand(command);
    }

    /**
     * 查询所有需要进行失效转移的任务实例.
     *
     * @param host 管理任务实例的工作节点主机地址 如果未指定 则忽略主机地址该限定条件.
     *
     * @return 需要进行失效转移的任务实例.
     */
    public List<TaskInstance> queryTaskInstanceNeedFailover(@Nullable final String host) {
        return this.taskInstance.queryByHostAndState(host, STATUS);
    }

    // -------------------------------------------------------------------------------------------------
    //  辅助方法 (指令交互)
    // -------------------------------------------------------------------------------------------------

    /**
     * 如果给定控制指令非空 则将该控制指令更新到指令集中.
     *
     * @param command 待更新到指令集中的控制指令.
     */
    private void createCommand(@Nullable final Command command) {
        if (command != null) {
            this.command.insert(command);
        }
    }

    /**
     * 从待处理的系统指令集中推出待执行的系统指令.
     *
     * @return 待执行的系统指令.
     */
    public Command findOneCommand() {
        return this.command.getOneToRun();
    }

    /**
     * 处理给定的系统控制指令.
     *
     * @param logger 用于记录处理过程中的日志.
     * @param host 用于管理工作流实例的主机地址.
     * @param restThreadNum 用于执行工作流实例的可用线程数.
     * @param command 待处理系统控制指令.
     *
     * @return 转译后的用于处理给定指令的工作流实例对象.
     */
    @Nullable
    @Transactional(rollbackFor = Exception.class)
    public WorkflowInstance handleCommand(
        final Logger logger,
        final String host,
        final int restThreadNum,
        final Command command
    ) {
        // 1. 从指令中解析或构建工作流实例.
        final WorkflowInstance instance = this.buildFromCommand(command);

        // 2. 如果无法构建出工作流实例 则判定此指令为错误指令.
        if (instance == null) {
            logger.error("Could not find the related workflow instance about the command.");

            this.moveToErrorCommand(command, "The workflow instance is null.");
            return null;
        }

        // 3. 填充工作流实例缺失的内容.
        instance.setHost(host);
        if (command.getScheduleTime() != null) {
            instance.setScheduleTime(command.getScheduleTime());
        }

        // 4. 根据指令类型填充相应的数据单元.
        final Map<String, String> params = JsonUtils.deserialize(
            command.getParams(),
            new TypeToken<HashMap<String, String>>(){}.getType()
        );
        ExecutionState state = ExecutionState.RUNNING;

        switch (command.getType()) {
            case RERUN:
                instance.setStartTime(new Date());
                instance.setEndTime(null);
                instance.setRunTimes(instance.getRunTimes() + 1);

                break;
            case RECOVER_FAILURE:
                // 5. 查询所有可以归类为失败的任务.
                final List<Integer> taskList = new ArrayList<>();

                taskList.addAll(this.taskInstance.queryByWorkflowInstanceIdAndState(
                    instance.getId(),
                    ExecutionState.FAILURE.ordinal()
                ));
                taskList.addAll(this.taskInstance.queryByWorkflowInstanceIdAndState(
                    instance.getId(),
                    ExecutionState.NEED_FAULT_TOLERANCE.ordinal()
                ));
                taskList.addAll(this.taskInstance.queryByWorkflowInstanceIdAndState(
                    instance.getId(),
                    ExecutionState.KILL.ordinal()
                ));

                for (final Integer taskId: taskList) {
                    final TaskInstance task = this.taskInstance.selectById(taskId);

                    task.setState(ExecutionState.SUBMITTED_SUCCESS);
                    this.saveTaskInstance(task);
                }

                // 6. 追加任务标识至指令参数中.
                params.put("taskIds", String.join(",", ArrayUtils.toStringArray(taskList.toArray())));

                // 7. 更新实例数据单元.
                instance.setCommandParams(JsonUtils.serializeToString(params));
                instance.setRunTimes(instance.getRunTimes() + 1);

                break;
            case RECOVER_SUSPENDED:
                // 8. 查询所有可以归类为暂停的任务.
                final List<Integer> tasks = new ArrayList<>();

                tasks.addAll(this.taskInstance.queryByWorkflowInstanceIdAndState(
                    instance.getId(),
                    ExecutionState.PAUSE.ordinal()
                ));
                tasks.addAll(this.taskInstance.queryByWorkflowInstanceIdAndState(
                    instance.getId(),
                    ExecutionState.KILL.ordinal()
                ));

                for (final Integer taskId: tasks) {
                    final TaskInstance task = this.taskInstance.selectById(taskId);

                    task.setState(ExecutionState.SUBMITTED_SUCCESS);
                    this.saveTaskInstance(task);
                }

                // 9. 追加任务标识至指令参数中.
                params.put("taskIds", String.join(",", ArrayUtils.toStringArray(tasks.toArray())));

                // 10. 更新实例数据单元.
                instance.setCommandParams(JsonUtils.serializeToString(params));
                instance.setRunTimes(instance.getRunTimes() + 1);

                break;
            case RECOVER_FAULT_TOLERANT:
                // 11. 更新实例中的恢复标识并回填原有实例状态.
                state = instance.getState();
                instance.setIsRecovery(Flag.YES);

                break;
            default:
                break;
        }

        instance.setState(state);

        // 12. 如果当前可用线程数不足 则更改系统指令为等待线程指令.
        if (restThreadNum == 0) {
            logger.info("There is not enough thread for the execution of the command {}.", command);

            instance.setState(ExecutionState.WAITING_THREAD);

            // 13. 如果当前指令非取消等待状态指令 则将当前指令追加到历史指令集中.
            if (command.getType() != CommandType.RECOVER_WAITING_THREAD) {
                instance.addHistoryCmd(command.getType());
            }

            // 14. 保存工作流实例并创建恢复线程指令.
            this.saveWorkflowInstance(instance);
            this.createRecoveryWaitingThreadCommand(command, instance);

            return null;
        }

        // 15. 更新工作流实例中的指令相关信息.
        instance.setCommandType(command.getType());
        instance.addHistoryCmd(command.getType());

        // 16. 保存工作流实例并删除原始指令.
        this.saveWorkflowInstance(instance);
        this.command.deleteById(command.getId());

        return instance;
    }

    /**
     * 从给定的控制指令中解析出相关联的工作流实例对象.
     *
     * @param command 待处理控制指令.
     *
     * @return 解析出的工作流实例对象.
     */
    @Nullable
    private WorkflowInstance buildFromCommand(final Command command) {
        // 1. 尝试获取该指令相关的工作流以构建工作流实例.
        Workflow flow = null;

        // 2. 如果指令中包含工作流信息 则查询该工作流以构建工作流实例.
        if (command.getWorkflowId() != 0) {
            flow = this.workflow.selectById(command.getWorkflowId());

            if (flow == null) {
                LOGGER.error("Could not find the related workflow {}.", command.getWorkflowId());

                return null;
            }
        }

        // 3. 如果无指令参数 则生成一个新的工作流实例.
        final Map<?, ?> params = JsonUtils.deserialize(command.getParams(), HashMap.class);

        if (params.isEmpty()) {
            return this.generateNewWorkflowInstance(flow, command);
        }

        // 4. 如果是工作流实例恢复类型的指令 则指令参数中必定包含工作流实例标识.
        int instanceId = 0;

        if (command.getType().name().toLowerCase(Locale.US).contains("recover")) {
            instanceId = Integer.parseInt(params.get("workflowInstanceId").toString());

            // 5. 确认工作流实例标识是否无效 (恢复等待线程类型指令无需指定特定工作流实例).
            if (instanceId == 0 && command.getType() != CommandType.RECOVER_WAITING_THREAD) {
                LOGGER.error("Could not extract the workflow instance id from the invalid command parameters.");

                return null;
            }
        }

        // 6. 查找或生成相应的工作流实例.
        final WorkflowInstance instance = instanceId == 0
            ? this.generateNewWorkflowInstance(flow, command)
            : this.workflowInstance.selectById(instanceId);

        // 7. 填充工作流定义到实例中.
        instance.setWorkflow(this.workflow.selectById(instance.getWorkflowId()));

        return instance;
    }

    /**
     * 从给定的工作流与控制指令及其参数中生成一个新的工作流实例对象.
     *
     * @param flow 构建该实例时所用的工作流定义.
     * @param command 用于填充实例中的控制指令相关内容.
     *
     * @return 生成的新的工作流实例对象.
     */
    private WorkflowInstance generateNewWorkflowInstance(final Workflow flow, final Command command) {
        // 1. 创建一个新的工作流实例对象.
        final WorkflowInstance instance = new WorkflowInstance(flow);

        // 2. 初始化该进程实例.
        instance.setPriority(command.getPriority());
        instance.setContent(flow.getContent());
        instance.setWorkflowId(command.getWorkflowId());
        instance.setStrategy(command.getStrategy());
        instance.setWarningType(command.getWarningType() == null ? WarningType.NONE : command.getWarningType());
        instance.setWarningGroupId(command.getWarningGroupId() == null ? 0 : command.getWarningGroupId());
        instance.setCommandType(command.getType());
        instance.setCommandParams(command.getParams());
        instance.setWorkerGroup(command.getWorkerGroup());
        instance.setMaxRetryCount(0);
        instance.setState(ExecutionState.RUNNING);
        instance.setStartTime(new Date());
        instance.setCommandStartTime(command.getStartTime());
        instance.setRunTimes(1);
        instance.setIsRecovery(Flag.NO);

        if (command.getScheduleTime() != null) {
            instance.setScheduleTime(command.getScheduleTime());
        }

        return instance;
    }

    /**
     * 使用给定的原始控制指令与工作流实例对象创建一个等待可用线程指令并存储至待处理指令集中.
     *
     * @param command 原始控制指令.
     * @param instance 从原始控制指令中解析出的工作流实例.
     */
    public void createRecoveryWaitingThreadCommand(@Nullable final Command command, final WorkflowInstance instance) {
        // 1. 构建命令参数.
        final Map<String, String> params = new HashMap<>();

        params.put("workflowInstanceId", String.valueOf(instance.getId()));

        // 3. 如果原始指令不存在 则以等待线程的状态退出该进程实例的执行.
        if (command == null) {
            this.saveCommand(new Command(
                CommandType.RECOVER_WAITING_THREAD,
                JsonUtils.serializeToString(params),
                instance.getWorkflowId(),
                instance.getPriority(),
                instance.getWorkerGroup(),
                instance.getStrategy(),
                instance.getWarningType(),
                instance.getWarningGroupId(),
                instance.getScheduleTime()
            ));

            return;
        }

        // 4. 如果原始指令为恢复等待线程 则仅需更新该命令的记录更新时间.
        if (command.getType() == CommandType.RECOVER_WAITING_THREAD) {
            command.setUpdateTime(new Date());

            this.saveCommand(command);
            return;
        }

        // 5. 移除原始指令并创建一个新的等待线程指令.
        this.command.deleteById(command.getId());

        command.setId(0);
        command.setType(CommandType.RECOVER_WAITING_THREAD);
        command.setParams(JsonUtils.serializeToString(params));
        command.setUpdateTime(new Date());
        command.setPriority(instance.getPriority());

        this.saveCommand(command);
    }

    /**
     * 存储给定的指令.
     *
     * @param command 需要存储的指令对象.
     */
    private void saveCommand(final Command command) {
        if (command.getId() == 0) {
            this.command.insert(command);

            return;
        }

        this.command.updateById(command);
    }

    /**
     * 将给定控制指令归类为执行错误的指令并删除原始指令.
     *
     * @param command 执行失败的原始指令.
     * @param cause 执行失败的原因.
     */
    @Transactional(rollbackFor = Exception.class)
    public void moveToErrorCommand(final Command command, final String cause) {
        this.errorCommand.insert(new ErrorCommand(command, cause));
        this.command.deleteById(command.getId());
    }

    /**
     * 将给定任务实例存储到数据库中并依据任务运行状态斟酌是否新建任务实例.
     *
     * @param task 待保存的任务实例.
     *
     * @return 如果存储成功则返回保存后的任务实例 否则返回 {@code null}.
     */
    @Nullable
    @Transactional(rollbackFor = Exception.class)
    public TaskInstance submitTask(final TaskInstance task) {
        // 1. 查询该任务实例所属工作流实例.
        final WorkflowInstance instance = this.findWorkflowInstanceById(task.getWorkflowInstanceId());

        // 2. 如果任务实例已经失败且工作流实例尚未准备暂停执行 则移除原有任务实例并创建一个新的任务实例.
        if (task.getState() == ExecutionState.FAILURE || task.getState() == ExecutionState.NEED_FAULT_TOLERANCE) {
            if (instance.getState() != ExecutionState.READY_STOP && instance.getState() != ExecutionState.READY_PAUSE) {
                this.taskInstance.deleteById(task.getId());

                // 3. 创建一个新的任务实例.
                task.setId(0);
                task.setHost(null);
                task.setStartTime(null);
                task.setEndTime(null);

                if (task.getState() != ExecutionState.NEED_FAULT_TOLERANCE) {
                    task.setRetryCount(task.getRetryCount() + 1);
                }
            }
        }

        // 4. 填充其他字段.
        task.setWorkflowInstancePriority(instance.getPriority());
        task.setSubmitTime(new Date());
        task.setState(this.getSubmitTaskState(task, instance));

        // 5. 如果更新失败 则反馈失败信息.
        if (this.saveTaskInstance(task)) {
            LOGGER.info("Succeed to submit the task {} to the database.", task.getName());

            return task;
        }

        LOGGER.info("Failed to submit the task {} to the database.", task.getName());

        return null;
    }

    /**
     * 确认是否需要更改给定任务实例的运行状态并获取更改后的状态.
     *
     * @param task 待确认运行状态的任务实例.
     * @param instance 该任务实例所属工作流实例.
     *
     * @return 给定任务更改后的运行状态.
     */
    private ExecutionState getSubmitTaskState(final TaskInstance task, final WorkflowInstance instance) {
        // 1. 如果任务实例状态正在执行或已经终止运行 则维持原状态.
        if (task.getState() == ExecutionState.RUNNING || task.getState() == ExecutionState.KILL) {
            return task.getState();
        }

        // 2. 如果工作流实例准备暂停 则判定任务实例为暂停状态.
        if (instance.getState() == ExecutionState.READY_PAUSE) {
            return ExecutionState.PAUSE;
        }

        // 3. 如果工作流实例准备停止或者工作流中存在执行失败的任务 则判定任务实例状态为停止.
        if (instance.getState() == ExecutionState.READY_STOP
            || this.findTaskInstancesByWorkflowInstanceId(instance.getId()).stream().anyMatch(item -> item.getState() == ExecutionState.FAILURE)) {
            return ExecutionState.KILL;
        }

        return ExecutionState.SUBMITTED_SUCCESS;
    }
}
